package com.ibm.cps.api;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.inject.Named;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;

import com.ibm.cps.kafka.JudgeTenantType;
import com.ibm.cps.kafka.KafkaTopicConstructorForMultiTenants;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.LocalConfig;

import kafka.admin.AdminUtils;
import kafka.utils.ZKStringSerializer;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.bson.types.ObjectId;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.jarxs.ParaUtility;
import com.ibm.cps.jarxs.TenantResource;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.model.AuthorizationManager;
import com.ibm.cps.model.ServletUtil;
import com.ibm.cps.object.Tenant;
import com.ibm.cps.servlet.Service;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

@Named
public class TenantBean implements TenantResource{


	private static Logger logger = LoggerFactory.getLogger(Tenant.class);
	@Autowired(required = true)
	private ApplicationContext applicationContext;
	
	@Autowired
	private ServiceBean service;
	
	private AuthorizationManager authManager;
	private JudgeTenantType judgeTenantType = new JudgeTenantType();
	private KafkaTopicConstructorForMultiTenants kafkaTopicConstructorForMultiTenants = new KafkaTopicConstructorForMultiTenants();
	private IPersistency persistency=null;
	public TenantBean() throws CPSException {
		persistency = MongoPersistency.getInstace();
		authManager = AuthorizationManager.getInstance();
	}


	@Override
	public Response PostTenant(String json, String auth) {
		try {

			
			if (!authManager.isAdmin(auth)) {
				ObjectNode node = JSONUtility.newObjectNode("message", "invalid access");
				node.put("code", 400);
				return Response.ok(node.toString(), MediaType.APPLICATION_JSON).build();
			}
			String tenantid = JSONUtility.getString(MessageFields.TENANTID, json);
			Pattern p = Pattern.compile(ParaUtility.regex);
			Matcher m = p.matcher(tenantid);
			if(!m.matches()){
				String ret = ServletUtil.constructReturnMsg(400, "invalid tentantid , should be this " + ParaUtility.regex);
				return Response.ok(ret,MediaType.APPLICATION_JSON).build();
			}

			String tenant_type = JSONUtility.getString(MessageFields.TENANT_TYPE, json);
			if(tenant_type == null || tenant_type.trim().equals("")){
				String ret = ServletUtil.constructReturnMsg(400, "invalid tenant type, " + tenant_type);
				return Response.ok(ret,MediaType.APPLICATION_JSON).build();
			}

			ObjectNode node = JSONUtility.newObjectNode();
			node.put(MessageFields.TENANTID, tenantid);
			node.put(MessageFields.TENANT_TYPE,tenant_type);
			node.put(MessageFields.ACCESS_TOKEN, new ObjectId().toString());
			//Persistence
			String ret = authManager.addCredential(node);
			CreateTopic(tenantid);
			//service
			service.PutService(tenantid,"{\"status\":\"start\"}",node.get(MessageFields.ACCESS_TOKEN).asText());
			
			return Response.ok(ret, MediaType.APPLICATION_JSON).build();
		} catch (CPSException e) {
			e.printStackTrace();
			return Response.ok(ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage()), MediaType.APPLICATION_JSON).build();
		}
	}


	private Boolean CreateTopic(String tenantid) throws CPSException {
		String normalTenant = MessageFields.NORMAL_TENANT;
		String dataTopic = new String();
		String processorTopic= new String();
		String processorDeleteTopic= new String();
		String datasourceAddTopic= new String();
		String datasourceDeleteTopic= new String();
		if(judgeTenantType.IsNormalTenants(tenantid))
		{
			
			if(persistency.getTenantStatus(normalTenant)==null)
			{

				dataTopic = kafkaTopicConstructorForMultiTenants.getMessageTopic(normalTenant);
				processorTopic = kafkaTopicConstructorForMultiTenants.getMetadataAddTopic(normalTenant);
				processorDeleteTopic = kafkaTopicConstructorForMultiTenants.getMetadataDeleteTopic(normalTenant);
				datasourceAddTopic = kafkaTopicConstructorForMultiTenants.getDataSourceAddTopic(normalTenant);
				datasourceDeleteTopic = kafkaTopicConstructorForMultiTenants.getDataSourceDeleteTopic(normalTenant);
				persistency.putTenantStatus(normalTenant,"exist");
			}else
				return true;
		}else{
			dataTopic = kafkaTopicConstructorForMultiTenants.getMessageTopic(tenantid);
			processorTopic = kafkaTopicConstructorForMultiTenants.getMetadataAddTopic(tenantid);
			processorDeleteTopic = kafkaTopicConstructorForMultiTenants.getMetadataDeleteTopic(tenantid);
			datasourceAddTopic = kafkaTopicConstructorForMultiTenants.getDataSourceAddTopic(tenantid);
			datasourceDeleteTopic = kafkaTopicConstructorForMultiTenants.getDataSourceDeleteTopic(tenantid);
		}

		ZkClient zkClient = new ZkClient(LocalConfig.ZOOKEEPER_HOST_PORT,10000 , 10000);
		zkClient.setZkSerializer(new ZkSerializer() {
			
			@Override
			public byte[] serialize(Object data) throws ZkMarshallingError {
				return ZKStringSerializer.serialize(data);
			}
			
			@Override
			public Object deserialize(byte[] bytes) throws ZkMarshallingError {
				return ZKStringSerializer.deserialize(bytes);
			}
		});
		AdminUtils.createTopic(zkClient, dataTopic, LocalConfig.KAFKA_PARTITION, LocalConfig.KAFKA_REPLICATIONFACTOR, new Properties());
		AdminUtils.createTopic(zkClient, processorTopic, LocalConfig.KAFKA_PARTITION, LocalConfig.KAFKA_REPLICATIONFACTOR, new Properties());
		AdminUtils.createTopic(zkClient, processorDeleteTopic, LocalConfig.KAFKA_PARTITION, LocalConfig.KAFKA_REPLICATIONFACTOR, new Properties());
		AdminUtils.createTopic(zkClient, datasourceAddTopic, LocalConfig.KAFKA_PARTITION, LocalConfig.KAFKA_REPLICATIONFACTOR, new Properties());
		AdminUtils.createTopic(zkClient, datasourceDeleteTopic, LocalConfig.KAFKA_PARTITION, LocalConfig.KAFKA_REPLICATIONFACTOR, new Properties());
		return true;
	}
	@Override
	public Response DeleteTenant(String tenantid, String auth) {
		try {
			if (!authManager.isAdmin(auth)) {

				ObjectNode node = JSONUtility.newObjectNode("message", "invalid access");
				node.put("code", 400);
				return Response.ok(node.toString(), MediaType.APPLICATION_JSON).build();
			}
			ObjectNode node = JSONUtility.newObjectNode(MessageFields.TENANTID, tenantid);
			String ret = authManager.deleteCredential(node);
			return Response.ok(ret, MediaType.APPLICATION_JSON).build();
		} catch (CPSException e) {
			e.printStackTrace();
			return Response.ok(ServletUtil.constructReturnMsg(200, e.getErrorMessage()), MediaType.APPLICATION_JSON).build();
		}
	}

	@Override
	public Response GetTenantInfo(String tenantId, String auth) {

		if (!authManager.isAdmin(auth)) {

			ObjectNode node = JSONUtility.newObjectNode("message", "invalid access");
			node.put("code", 400);
			return Response.ok(node.toString(), MediaType.APPLICATION_JSON).build();
		}
		String ret = authManager.getCredential(tenantId);
		return Response.ok(ret, MediaType.APPLICATION_JSON).build();

	}
}

